iT邦幫忙

2026 iThome 鐵人賽

DAY 22
0
Software Development

從零打造邊緣運算閘道器:Raspberry Pi 與 Linux 底層軟硬整合實戰系列 第 22

Day 22 - 優先權設計:如何確保「致命錯誤」的紅燈永遠不會被「待機」的綠燈覆蓋?

  • 分享至 

  • xImage
  •  

昨天的文章中,我們成功建立了一個統一發號施令的 LEDController,並實作了基於 cancel_flag 的搶佔機制。現在,只要呼叫 request_animation(),新的動畫就能立刻中斷並取代舊的動畫。

但這引發了一個嚴重的架構漏洞。

[問題情境] 愚蠢的搶佔機制

想像 EdgeNode 的運作流程:

  1. 系統偵測到磁碟空間耗盡,呼叫 request_animation(solid_red) 顯示致命錯誤紅燈。
  2. 三秒後,一個負責定時檢查網路連線的背景執行緒完成了檢查,依據它的邏輯,網路正常就呼叫 request_animation(breathe_green)

結果:紅燈只亮了三秒,就被不知情的綠燈覆蓋了!

維修人員看到綠燈,以為設備運作正常,殊不知設備早就因為磁碟耗盡而癱瘓。這在醫療儀器或工控領域是絕對不可原諒的。

[錯誤嘗試] 到處塞滿 if 判斷式

直覺的做法是在每個發送請求的地方加上狀態判斷:

if current_state != ERROR_STATE:
    led_controller.request_animation(breathe_green)

但這會讓整個專案的程式碼變得極度混亂(Spaghetti Code)。狀態機不應該散落在各個業務邏輯中,這違反了我們「解耦」的初衷。

[底層原理] 優先權 (Priority) 與世代計數器 (Generation)

解決方案是為每個任務賦予一個優先權 (Priority)
LEDController 必須記住當前動畫的優先級別,當新任務進來時,只有當新任務的優先級 大於或等於 當前任務時,才允許搶佔。

同時,我們面臨 Python threading 中的一個痛點:我們很難安全且立即地「殺死」一個執行緒。前一篇我們用了一把全域的 _cancel_flag,但如果頻繁切換動畫,這個單一旗標可能會引發時序上的 Race Condition(舊執行緒把新執行緒的旗標給清掉了)。

為此,我們引入資料庫設計常見的概念:世代計數器 (Generation Counter) 或稱 Job ID。

[最終解決方案] 嚴謹的優先權引擎

讓我們將 LEDController 升級為最終的企業級版本:

import threading
import time

class LEDController:
    # 定義優先權 (數字越大越優先)
    PRIO_IDLE = 1
    PRIO_SYNCING = 3
    PRIO_ERROR = 5

    def __init__(self, hardware):
        self.hw = hardware
        self._current_prio = 0
        self._current_job_id = 0
        self._lock = threading.Lock()
        
    def _run_anim(self, anim_func, job_id):
        # check_cancel 函式:判斷自己這個世代是否已經過期
        def check_cancel():
            with self._lock:
                return self._current_job_id != job_id

        # 執行實際動畫
        anim_func(self.hw, check_cancel)
        
    def request(self, anim_func, priority):
        with self._lock:
            # 核心邏輯:低於當前優先權的請求直接被忽略
            if priority < self._current_prio:
                print(f"[LED] 拒絕低優先級請求 ({priority} < {self._current_prio})")
                return False
                
            # 更新狀態與世代交替
            self._current_prio = priority
            self._current_job_id += 1
            new_job_id = self._current_job_id
            
        print(f"[LED] 啟動新動畫 (Prio: {priority}, Job: {new_job_id})")
        
        # 啟動新執行緒
        threading.Thread(
            target=self._run_anim, 
            args=(anim_func, new_job_id),
            daemon=True
        ).start()
        
        return True

    def clear_error(self):
        """專門用來解除高優先級鎖定的方法"""
        with self._lock:
            self._current_prio = 0
        self.request(breathe_green, self.PRIO_IDLE)

[!TIP]
為何使用 Job ID?
在這種設計中,我們不需要 join() 等待舊執行緒死掉。只要 _current_job_id 遞增,舊執行緒在下一次呼叫 check_cancel() 時,就會發現 自己的 job_id != 系統最新的 job_id,然後默默退出。這種孤兒進程自動消亡的模式,極大地提升了系統反應速度。

實際應用

現在,當磁碟耗盡時,我們發送:
led_controller.request(solid_red, LEDController.PRIO_ERROR)

當網路定時檢查完成,它發送:
led_controller.request(breathe_green, LEDController.PRIO_IDLE)

LEDController 內部的 self._lock 會發現 PRIO_IDLE (1) < PRIO_ERROR (5),直接退回請求。紅燈得以堅挺地亮著,直到系統管理員前來排除障礙並呼叫 clear_error()

我們不僅解決了硬體控制的多執行緒問題,還為它建構了一套堅不可摧的狀態機防禦機制。

這就是軟體工程的魅力所在。

明天,我們將探討另一個 Python 開發者的噩夢:time.sleep() 帶來的阻塞與按鍵防彈跳 (Debounce) 的處理策略。Day 23 見!


上一篇
Day 21 - 軟即時架構 (Soft Real-Time):自己當交通警察,打造搶佔式任務排程
下一篇
Day 23 - 如何優雅地殺死 Python Thread?世代計數器 (Generation Counter) 的妙用
系列文
從零打造邊緣運算閘道器:Raspberry Pi 與 Linux 底層軟硬整合實戰26
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言